package com.joey.concurrency.example.queue;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author hyy
 * @description
 * @date 2019-03-20 11:14
 **/
@Slf4j
public class SynchronousQueueExample {
    /**
     * SynchronousQueue没有容量，是无缓冲等待队列，是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。
     * 拥有公平（FIFO）和非公平(LIFO)策略，非公平侧罗会导致一些数据永远无法被消费的情况？
     * 使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作。
     *
     * LinkedBlockingQueue是一个无界缓存等待队列。当前执行的线程数量达到corePoolSize的数量时，
     * 剩余的元素会在阻塞队列里等待。（所以在使用此阻塞队列时maximumPoolSizes就相当于无效了），
     * 每个线程完全独立于其他线程。生产者和消费者使用独立的锁来控制数据的同步，即在高并发的情况
     * 下可以并行操作队列中的数据。
     *
     * ArrayBlockingQueue是一个有界缓存等待队列，可以指定缓存队列的大小，当正在执行的线程数等于corePoolSize时，
     * 多余的元素缓存在ArrayBlockingQueue队列中等待有空闲的线程时继续执行，当ArrayBlockingQueue已满时，
     * 加入ArrayBlockingQueue失败，会开启新的线程去执行，当线程数已经达到最大的maximumPoolSizes时，
     * 再有新的元素尝试加入ArrayBlockingQueue时会报错。
     */
    public static void main(String[] args) throws InterruptedException {

        SynchronousQueue<Integer> synchronousQueue = new SynchronousQueue<Integer>(true);

        Thread putThread = new Thread(new Runnable() {
            @Override
            public void run() {
                log.info("put start");
                try {
                    for (int i = 0; i < 200; i++) {
                        log.info("put {}",i);
                        synchronousQueue.put(i);
                    }
//                    Thread.sleep(5 * 1000);

                    System.out.println(synchronousQueue.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                log.info("put end");
            }
        });



        Thread takeThread = new Thread(new Runnable() {
            @Override
            public void run() {
                log.info("take start");
                try {
                    for (int i = 0; i < 100; i++) {
                        log.info("take wait 。。。");
                        Thread.sleep(2000);
                        log.info("take from {}",synchronousQueue.take());
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("take end");
            }
        });

        putThread.start();
//        Thread.sleep(5000);
        takeThread.start();
    }


}




    